package ranks.tasks.calc;

import java.io.IOException;
import java.util.Iterator;

import mapred.example.base.TextCombiner;

import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reporter;

import ranks.base.Const;
import ranks.base.Data;

public class Calc_Combiner extends TextCombiner {

	@Override
	protected void combine(Text key, Iterator<Text> values,
			OutputCollector<Text, Text> output, Reporter reporter)
			throws IOException {
		Data sum = new Data();
		while (values.hasNext()) {
			Text value = values.next();
			String text = value.toString().trim();
			if (text.startsWith(Const.RANK)) {
				Data data = new Data(text.substring(Const.RANK.length()));
				sum.add(data);
			} else {
				output.collect(key, value);
			}
		}
		output.collect(key, new Text(Const.RANK + sum.toString()));
	}

	@Override
	public void setUp() {
	}

}
